package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MykafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName DwdTradePaymentAdd
 * @Description TODO 支付成功人数
 * @Author AFei
 * @Date 2022/9/6 9:53
 * @Version 1.0
 */
public class DwdTradePaymentAdd {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(MykafkaUtil.getTopicDDL("dwd_trade_payment_add"));

        // TODO 从业务数据中筛选出支付成功人数
        Table paymentAdd = tableEnv.sqlQuery("select\n" +
                "        data['id'] id,\n" +
                "        data['out_trade_no'] out_trade_no,\n" +
                "        data['order_id'] order_id,\n" +
                "        data['trade_body'] trade_body,\n" +
                "        data['payment_type'] payment_type,\n" +
                "        data['payment_status'] payment_status,\n" +
                "        ts,\n" +
                "        proc_time\n" +
                "        from topic_db where `table`='payment_info' ");
        tableEnv.createTemporaryView("payment_add",paymentAdd);
        //tableEnv.executeSql("select * from payment_add").print();

        // TODO 将关联的结果写到kafka主题中
        // 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("CREATE TABLE dwd_trade_payment_add ( \n" +
                "                  id string, \n" +
                "                  out_trade_no string, \n" +
                "                  order_id string, \n" +
                "                  trade_body string, \n" +
                "                  payment_type string, \n" +
                "                  payment_status string,\n" +
                "                  ts string, \n" +
                "                  proc_time TIMESTAMP_LTZ(3), \n" +
                "                  PRIMARY KEY (id) NOT ENFORCED \n" +
                "                )" + MykafkaUtil.getUpsert("dwd_trade_payment_add"));
        tableEnv.executeSql("insert into dwd_trade_payment_add select * from payment_add");
    }
}
